package com.waleipt.api.es.tcptransport;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Supplier;

import org.elasticsearch.Version;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.CheckedBiConsumer;
import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
import org.elasticsearch.common.network.NetworkService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.transport.TransportAddress;
import org.elasticsearch.common.util.BigArrays;
import org.elasticsearch.indices.breaker.NoneCircuitBreakerService;
import org.elasticsearch.node.Node;
import org.elasticsearch.tasks.TaskManager;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.ConnectionProfile;
import org.elasticsearch.transport.RequestHandlerRegistry;
import org.elasticsearch.transport.Transport;
import org.elasticsearch.transport.Transport.Connection;
import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.transport.TransportRequestHandler;
import org.elasticsearch.transport.TransportRequestOptions;
import org.elasticsearch.transport.TransportResponse;
import org.elasticsearch.transport.TransportResponseHandler;
import org.elasticsearch.transport.TransportResponseOptions;
import org.elasticsearch.transport.TransportServiceAdapter;
import org.elasticsearch.transport.TransportSettings;
import org.elasticsearch.transport.netty4.Netty4Transport;

public class Client_Node {
	
	Map<Long,TransportResponseHandler> clientHandler = new HashMap<>() ;
	Netty4Transport _client ;
	Transport.Connection connection ;
	
	public Client_Node( int port ) {
		
		ThreadPool threadPool = new ThreadPool(Settings.builder()
				.put(Node.NODE_NAME_SETTING.getKey(), this.getClass().getName() )
				.build()) ;
		
		Settings client_settings = Settings.builder()
				//					.put(TcpTransport.PING_SCHEDULE.getKey(), "1000ms" )
									.put(TransportSettings.BIND_HOST.getKey(), "10.100.33.155" )
									.put(TransportSettings.PUBLISH_HOST.getKey(), "10.100.33.155" )
									.put(TransportSettings.PORT.getKey(), port )
									.build() ;
        _client = new Netty4Transport(client_settings, 
									threadPool, 
									new NetworkService( client_settings, Collections.emptyList()), 
									new BigArrays(Settings.EMPTY, null), 
									new NamedWriteableRegistry(Collections.emptyList()), 
									new NoneCircuitBreakerService()) ;
        _client.transportServiceAdapter( new ClientTransportServiceAdapter() );
		
	}
	public void start()
	{
		 _client.start();
	}
	
	public void connectToNode( String address, int port ) throws UnknownHostException
	{
		DiscoveryNode node = new DiscoveryNode( "nodeId_1", new TransportAddress( InetAddress.getByName(address), port), Version.CURRENT);
		// 连接Node,发送连接node请求报文
		_client.connectToNode(node, null, new CheckedBiConsumer<Transport.Connection, ConnectionProfile, IOException>() {
			
			@Override
			public void accept(Connection t, ConnectionProfile u) throws IOException {
				
			}
		} );
		connection = _client.getConnection(node);
	}
	
	public void sendRequest( long requestId, String action, TransportRequest request, TransportResponseHandler responseHandler ) throws IOException, TransportException 
	{
		connection.sendRequest( requestId, action, request, TransportRequestOptions.EMPTY );
		clientHandler.put( requestId, responseHandler) ;
	}
	
	
	
	
	class ClientTransportServiceAdapter implements TransportServiceAdapter
	{
		volatile Map<String, RequestHandlerRegistry> requestHandlers = new HashMap<>() ;
		
		public ClientTransportServiceAdapter() {
		}
		
		public <Request extends TransportRequest> RequestHandlerRegistry regHadlerRequest( String action, Supplier<Request> requestFactory, TransportRequestHandler requsetHandler )
		{
			TaskManager taskManager = new TaskManager( Settings.builder().build() ) ;
			String executor = ThreadPool.Names.SAME ;
			boolean forceExecution = false ;
			boolean canTripCircuitBreaker = false ;
			RequestHandlerRegistry reg = new RequestHandlerRegistry<>( action, requestFactory, taskManager, 
					requsetHandler, 
					executor, forceExecution, canTripCircuitBreaker) ;
			return requestHandlers.put( action, reg) ;
		}
		
		@Override
		public void addBytesReceived(long size) {
			// TODO Auto-generated method stub
			
		}
		
		@Override
		public void addBytesSent(long size) {
			// TODO Auto-generated method stub
			
		}
		
		@Override
		public void onRequestSent(DiscoveryNode node, long requestId, String action, TransportRequest request,
				TransportRequestOptions options) {
			// TODO Auto-generated method stub
			
		}
		
		@Override
		public void onResponseSent(long requestId, String action, TransportResponse response,
				TransportResponseOptions options) {
		}
		
		@Override
		public void onResponseSent(long requestId, String action, Exception e) {
			// TODO Auto-generated method stub
			
		}
		
		// 接受响应处理客户端处理
		@Override
		public TransportResponseHandler onResponseReceived(long requestId) {
			return clientHandler.remove(requestId) ;
		}
		
		@Override
		public void onRequestReceived(long requestId, String action) {
			// TODO Auto-generated method stub
			
		}
		
		// 接受请求处理服务端处理
		@Override
		public RequestHandlerRegistry getRequestHandler(String action) {
			return null;
		}
		
	}
}
